package com.corn.flink.lesson3;

import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentBuilder;

import java.util.ArrayList;
import java.util.List;

/**
 * Es Sink Operator
 *
 * @author JimWu
 * @date 2023/2/28 18:03
 **/
public class FlinkEsSinkOperatorDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Person> ds = env.fromElements(
                new Person("alice", 15, "cq"),
                new Person("jack", 18, "cq"),
                new Person("rose", 20, "sh"),
                new Person("mick", 42, "gz"),
                new Person("white", 12, "cq"),
                new Person("maria", 32, "sz"),
                new Person("candy", 56, "bj"),
                new Person("jerry", 28, "wh")
        );

        List<HttpHost> hosts = new ArrayList<>();
        hosts.add(new HttpHost("localhost", 9200));
        ds.addSink(new ElasticsearchSink.Builder<Person>(hosts, new ElasticsearchSinkFunction<Person>() {
            @Override
            public void process(Person person, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
                IndexRequest indexRequest = Requests.indexRequest()
                        .index("person")
                        .source("data",person.toString());
                requestIndexer.add(indexRequest);
            }
        }).build());

        env.execute();
    }
}
